package com.meihua.code.thread;

import com.qingqing.common.async.qqmq.QqmqMsgConvert;
import com.qingqing.common.async.rabbit.AsyncEventNotifyRabbitMsgEntity.AsyncEventNotifyRabbitMsgBody;
import com.qingqing.common.rabbit.bean.RabbitMessageBody;
import com.qingqing.common.rabbit.bean.RabbitMessageEntity;
import com.qingqing.common.rabbit.bean.RabbitMessageHeader;
import com.qingqing.common.rabbit.bean.RabbitMsgEntityRouteWarpper;
import com.qingqing.common.util.JsonUtil;
import com.qingqing.common.web.aspect.log.InheritedGuid;
import com.qingqing.common.web.aspect.log.InheritedGuid.Ops;
import com.qingqing.common.web.util.GuidUtil;
import com.qingqing.qqmq.base.producer.builder.MessageBuilder;
import com.qingqing.qqmq.service.QQMQProduceService;
import com.qingqing.qqmq.thrift.Result;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * qqmq 消息发送者
 */
public class AsyncQqmqRouteSenderV2 {

    private static final Logger logger = LoggerFactory.getLogger(AsyncQqmqRouteSenderV2.class);

    private final LinkedBlockingQueue<RabbitMsgEntityRouteWarpper> iq = new LinkedBlockingQueue<RabbitMsgEntityRouteWarpper>(10000);
    private ExecutorService service;
    private static final int DEFAULT_THREAD_COUNT = 1;

    private String senderName = "";
    private Integer threadCount = DEFAULT_THREAD_COUNT;

    private final QQMQProduceService qqmqProduceService;
    private final String topicName;
    private QqmqMsgConvert convert;


    public AsyncQqmqRouteSenderV2(QQMQProduceService qqmqProduceService,
                                  String topicName,
                                  QqmqMsgConvert qqmqMsgConvert) {
        this.qqmqProduceService = qqmqProduceService;
        this.topicName = topicName;
        this.convert = qqmqMsgConvert;
    }

    public void start() {
        stop();
        service = new ThreadPoolExecutor(1, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(10000));
        for( int j = 0; j < threadCount; j++ ) {
            final int finalJ = j;
            service.execute(() -> {
                logger.info("qqmq async message thread has started, senderName:{}-{}", senderName, finalJ);

                while( true ) {
                    RabbitMsgEntityRouteWarpper wrapper = null;
                    try {
                        wrapper = iq.take();
                    } catch (InterruptedException e) {
                        logger.info("qqmq async message thread stop");
                        break;
                    }

                    if( wrapper == null ) {
                        continue;
                    }

                    GuidUtil.tryUseGuid(wrapper);
                    convertAndSenderMessage(wrapper);
                }
            });
        }

    }

    // send message
    private void convertAndSenderMessage(RabbitMsgEntityRouteWarpper wrapper){
        if(true){
            //   logger.info(Thread.currentThread().getName()+"模拟发送");
            try {
                Thread.sleep(10);
            }catch (Exception e){

            }
            return;
        }
        try {
            if( logger.isDebugEnabled() ) {
                logger.debug("send qqmq msg: {}", JsonUtil.getJsonFromObject(wrapper));
            }
            MessageBuilder messageBuilder = qqmqProduceService.messageBuilder()
                    .setTopic(topicName)
                    .setBody(convertMessage(wrapper))
                    .setTags(convertTags(wrapper.getTags()));

            if(!StringUtils.isBlank(wrapper.getRouteKey())){
                messageBuilder.setRoutingKey(wrapper.getRouteKey());
            }

            Result r = messageBuilder.send();
            if( r.code != 0 ) {
                logger.error("send qqmq msg error, ignore it. send resultCode:{}, resultKey:{}, resultMsg:{}. msg:{}",
                        r.getCode(), r.getKey(), r.getMsg(), JsonUtil.getJsonFromObject(wrapper));
            }
        } catch (RuntimeException e) {
            logger.error("send qqmq msg error, ignore it. msg:{}", JsonUtil.getJsonFromObject(wrapper), e);
        }
    }

    private String convertMessage(RabbitMsgEntityRouteWarpper wrapper){
        RabbitMessageBody body = wrapper.getEntity();
        RabbitMessageHeader header = convert.formMsgHeader(body.getVersion());
        RabbitMessageEntity entity = new RabbitMessageEntity(header, body);
        return JsonUtil.getJsonFromObject(entity);
    }

    public void stop() {
        if(service != null){
            logger.info("try to stop AsyncQqmqSender.");
            service.shutdownNow();
            logger.info("stop AsyncQqmqSender suc.");
        }
    }

    @InheritedGuid(ops = Ops.save)
    public void sendMsg(RabbitMsgEntityRouteWarpper msg) {
        try {
            sendMsgToQueue(msg);
        } catch (RuntimeException ignore) {
            logger.error("send qqmq msg to queue error. msg:{}", JsonUtil.getJsonFromObject(msg), ignore);
        }
    }

    private void sendMsgToQueue(RabbitMsgEntityRouteWarpper msg) {
        if(!iq.offer(msg)) {

            if (msg.getEntity()!=null && ! (msg.getEntity() instanceof AsyncEventNotifyRabbitMsgBody)){
                convertAndSenderMessage(msg);
                // logger.warn("mq msg queue is full. sender mq message sync :{}", JsonUtil.getJsonFromObject(task.msg));
            }else {
                // logger.error("mq msg queue is full. sender mq message discard policy :{}", JsonUtil.getJsonFromObject(task.msg));
            }
           // logger.error("qqmq msg queue is full. sender Qqmq message sync :{}", JsonUtil.getJsonFromObject(msg));
        } else if( logger.isTraceEnabled() ){
            logger.trace("pushed into queue suc. msg:{}", JsonUtil.getJsonFromObject(msg));
        }
    }

    private static String [] convertTags(String... values){
        if(values!=null){
            if (values.length<=0){
                return null;
            }
            for(String value:values){
                if(StringUtils.isNotBlank(value)){
                    return values;
                }
            }
            return null;
        }
        return null;
    }

    public void setConvert(QqmqMsgConvert convert) {
        this.convert = convert;
    }

    public void setThreadCount(Integer threadCount) {
        this.threadCount = threadCount;
    }

    public void setSenderName(String senderName) {
        this.senderName = senderName;
    }

    public static void main(String[] args) {
        AsyncQqmqRouteSenderV2 asyncQqmqRouteSenderV2 = new AsyncQqmqRouteSenderV2(null,
                "tst",null);
        asyncQqmqRouteSenderV2.start();

        long l = System.currentTimeMillis();

        for (int i=0;i<20000;i++){
            try {
                Thread.sleep(3);
                RabbitMsgEntityRouteWarpper warpper= new RabbitMsgEntityRouteWarpper();
                AsyncEventNotifyRabbitMsgBody body = new AsyncEventNotifyRabbitMsgBody();
                warpper.setMsgBody(body);
                body.setRef_id("1");
                asyncQqmqRouteSenderV2.sendMsg(warpper);
            }catch (Exception e){

            }

        }
        asyncQqmqRouteSenderV2.stop();
        logger.info("done 耗时："+(System.currentTimeMillis()-l));
    }

}
